package hydrap2p.manager;

import hydrap2p.Service;
import hydrap2p.ServiceLocation;
import hydrap2p.helper.ThriftServerThread;
import hydrap2p.helper.ThriftUtils;
import hydrap2p.helper.Utility;
import hydrap2p.modulemanager.ModuleInfo;
import hydrap2p.modulemanager.ModuleManager;
import hydrap2p.modulemanager.ModuleManagerServiceListener;
import hydrap2p.servicemanager.NameTaken;
import hydrap2p.servicemanager.ServiceListener;
import hydrap2p.servicemanager.ServiceManager;
import hydrap2p.speedlistener.SpeedListener;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.BindException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;

import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TServerTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

public class ManagerImpl implements ServiceManager.Iface, ModuleManager.Iface {
	
	private Map<String,ServiceLocation> locations;
	private ModuleManagerImpl confManager;
	private ServiceListeners listeners;
	
	private Timer pinger;
	private PingTimer pingtimer;
	private final int pingerdelay=5000;
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		int port=-1;
		
		try {
			port=Utility.getManagerPort();
		} catch (FileNotFoundException e) {
			System.err.println("Could not load configuration: "+e.getLocalizedMessage());
			System.err.println("Please provide a valid configuration of where to locate the Manager, and restart this program");
			return;
		}

		try {
			new ManagerImpl().run(port);
		} catch (BindException e) {
			System.err.println("Could not start listening, check that nothing is blocking the port");
		} catch (IOException e) {
			e.printStackTrace();
		}

	}
	
	public ManagerImpl()
	{
		locations=new HashMap<String,ServiceLocation>();
		confManager=new ModuleManagerImpl(this);
		listeners=new ServiceListeners();
	}
	
	private class PingTimer extends TimerTask {

		private TimeoutChecker timeouts;
		public PingTimer()
		{
			timeouts=new TimeoutChecker();
			timeouts.start();
		}
		
		@Override
		public void run() {
			// check the locations
			Map<String,ServiceLocation> curlocations=null;
			synchronized(locations)
			{
				curlocations=new HashMap<String,ServiceLocation>(locations);
			}
			for (String name:curlocations.keySet())
			{
				ServiceLocation there=curlocations.get(name);
				TSocket transport=new TSocket(there.address,there.port);
				timeouts.setCurrentConnection(transport);
				try {
					transport.open();
					Service.Client connection=new Service.Client(new TBinaryProtocol(transport));
					connection.ping();
				} catch (TException e) {
					System.err.println("Failed to ping "+name+", removing");
					removeLocation(name);
				}
				timeouts.finishedCurrentConnect();
			}
		}
		
		public void shutdown()
		{
			timeouts.shutdown();
		}
	}
	
	private void run(int port) throws IOException
	{
		System.err.println("Starting server");
		
		// bind to the socket
		ServerSocket socket = new ServerSocket();
		socket.bind(new InetSocketAddress("localhost",port));
		TServerTransport tsocket = new TServerSocket(socket);
		TThreadPoolServer.Args args = new TThreadPoolServer.Args(tsocket);
		args.processor(new ServiceManager.Processor(this));
		TThreadPoolServer mine = new TThreadPoolServer(args);
		System.err.println("Bound server to port "+port);
		
		// startup ModuleManager
		ThriftServerThread modulethread;
		modulethread=null;
		try {
			modulethread=ThriftUtils.initThriftServer(new ModuleManager.Processor(this), null);
			modulethread.start();
		} catch (BindException e) {
			throw e;
		} catch (IOException e) {
			throw e;
		} catch (NameTaken e) {
			throw new IOException("Couldn't register at ServiceManager");
		}
		try {
			setServiceLocation("ModuleManager", modulethread.getLocation());
		} catch (NameTaken e) {
			throw new IOException("Couldn't register at ServiceManager");
		} catch (TException e) {
			throw new IOException("Couldn't register at ServiceManager");
		}
		
		// start pinging our known services
		pinger=new Timer();
		pingtimer=new PingTimer();
		pinger.schedule(pingtimer, pingerdelay, pingerdelay);
		
		// Port has been bound, we can start other programs and they will just block until the TServer starts
		for (String module:confManager.getModuleNames())
			if (confManager.isAutostart(module))
				confManager.startModule(module);
		
		mine.serve();
		System.err.println("Stopping");
		shutdown();
	}
	
	private void removeLocation(String name)
	{
		synchronized (locations)
		{
			locations.remove(name);
		}
		confManager.serviceStopped(name);
		
		// notify listeners
		listeners.delService(name);
	}

	@Override
	public void ping() throws TException {
		return;
	}

	@Override
	public void delServiceLocation(String name) throws TException {
		removeLocation(name);
		
	}

	@Override
	public Map<String, ServiceLocation> getAllServiceLocations()
			throws TException {
		return locations;
	}

	@Override
	public List<ModuleInfo> getModuleList() throws TException {
		LinkedList<ModuleInfo> output=new LinkedList<ModuleInfo>();
		for (String name:confManager.getModuleNames())
		{
			ModuleInfo test=confManager.getModuleInfo(name);
			if (test!=null)
			{
				output.add(test);
				//System.out.println(test);
			}
		}
		return output;
	}

	@Override
	public ServiceLocation getServiceLocation(String name) throws TException {
		ServiceLocation ret;
		synchronized(locations)
		{
			ret=locations.get(name);
		}
		return ret;
	}

	@Override
	public void setModuleAutostart(String name, String server, boolean autostart)
			throws TException {
		confManager.setAutostart(name, autostart);
		
	}

	@Override
	public void setServiceLocation(String name, ServiceLocation location)
			throws NameTaken, TException {
		synchronized(locations){
			if (locations.containsKey(name))
				throw new NameTaken(name,locations.get(name));
			else
				locations.put(name, location);	
		}
		confManager.serviceStarted(name);
		System.out.println("Stored the location for "+name+" at "+location.address+":"+location.port);
		
		// notify listeners about new service
		listeners.newService(name, location);
	}

	@Override
	public boolean subscribe(ServiceLocation listener) throws TException {
		return listeners.addListener(listener, locations);
	}

	@Override
	public void unsubscribe(ServiceLocation listener) throws TException {
		listeners.removeListener(listener);		
	}

	@Override
	public void startModule(String name, String server) throws TException {
		confManager.startModule(name);
		
	}

	@Override
	public void stopModule(String name, String server) throws TException {
		confManager.stopModule(name);
		
	}

	@Override
	public List<String> supportedInterfaces() throws TException {
		LinkedList<String> temp=new LinkedList<String>();
		temp.add("Manager");
		return temp;
	}

	@Override
	public void shutdown() {
		pingtimer.shutdown();
		pinger.cancel();
		listeners.shutdown();
		confManager.stopAllModules();
		System.exit(0);
	}


}
